/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.samples.nbNative;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Enumeration;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.narada.event.EventProperties;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventException;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.service.replay.ReplayEvent;
import cgl.narada.service.replay.ReplayRequest;
import cgl.narada.service.replay.ReplayResponse;
import cgl.narada.service.replay.ReplayServiceListener;
/**
   @author Shrideep Pallickara and Hasan Bulut
   $Date: 2006/05/05 17:32:15 $
   $Revision: 1.3 $
 */

public class TDSClient 
  implements NBEventListener, NBRecoveryListener, ReplayServiceListener {
  private int entityId;
  private ClientService clientService;
  private String moduleName= "TDSClient: ";
  private EventConsumer consumer;
  private EventProducer producer;
  private ProducerConstraints producerConstraints;
  private Profile profile;
  private int templateID = 3142;
  private String synopsis = "Media/Data/1000";


  public TDSClient(int entityId) {
    try {
      this.entityId = entityId;
      //SessionService.setServiceConfigurationLocation("D:/NaradaSources/config/ServiceConfiguration.txt");
      clientService = SessionService.getClientService(entityId);
    } catch (ServiceException serEx) {
      System.out.println(serEx);
    }
  }
  
  public void 
  initializeBrokerCommunications(Properties props, String commType) 
    throws ServiceException{
    clientService.initializeBrokerCommunications(props, commType);
  }
  
  public void closeBrokerConnection() throws ServiceException {
    clientService.closeBrokerConnection();
    clientService.terminateServices();
  }


  public void initializeProducer() throws ServiceException {
    producer = clientService.createEventProducer();

    producer.generateEventIdentifier(true);
    producer.setTemplateId(templateID);
    producer.setDisableTimestamp(false);
    producer.setHighResolutionTimestamp(true);
    producer.setSuppressRedistributionToSource(true);

    TemplateInfo templateInfo = 
      clientService.createTemplateInfo(templateID, 
				       TemplateProfileAndSynopsisTypes.STRING,
				       synopsis);
    producerConstraints = 
      producer.createProducerConstraints(templateInfo);
    producerConstraints.setSendReliably();
    long recoveryId = producer.recover(templateID, this);
    System.out.println(moduleName + "Assigned recovery id = [" +
		       recoveryId + "] \n\n");
  }
  
  public void initializeConsumer() throws ServiceException {
    profile = 
      clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
				  synopsis);
    consumer = clientService.createEventConsumer(this);
    ConsumerConstraints constraints = 
      consumer.createConsumerConstraints(profile);
    constraints.setReceiveReliably(templateID);
    // constraints.setReceiveInTimeOrder();
    //constraints.setReceiveAfterTimeSpacing();
    consumer.subscribeTo(profile, constraints);
    long recoveryId = consumer.recover(templateID, this);
  }
  
  public void initializeConsumerWithTDS() throws ServiceException {
    profile = 
      clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
				  synopsis);
    consumer = clientService.createEventConsumer(this);
    ConsumerConstraints constraints = 
      consumer.createConsumerConstraints(profile);
    //constraints.setReceiveReliably(templateID);
    constraints.setReceiveInTimeOrder();
    constraints.setReceiveAfterTimeSpacing();
    consumer.subscribeTo(profile, constraints);
    //long recoveryId = consumer.recover(templateID, this);
  }

  public void unsubscribe() throws ServiceException { 
    consumer.unSubscribe(profile);
  }

  public void initiateReplay(long start, long stop) 
    throws ServiceException {
    ReplayRequest replayRequest = 
      clientService.createReplayRequest(templateID, start , stop );
    consumer.initiateReplay(replayRequest, this); 
  }

  public void onReplay(ReplayEvent replayEvent) {
    System.out.println(moduleName + "Received Replay Event");
    onEvent(replayEvent.getNBEvent());
  }

   public void onReplayResponse(ReplayResponse replayResponse) {
    System.out.println("\n\n\n\n" + moduleName + replayResponse + "\n\n");
  }

  public void publishEvent(String stringMsg) 
    throws ServiceException, NBEventException {
    if (stringMsg.equals("")) {
      stringMsg += System.currentTimeMillis();
    }
   
    NBEvent nbEvent = 
      producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			     synopsis, stringMsg.getBytes());
    producer.publishEvent(nbEvent, producerConstraints);
  }
  

  
  /** Upon completion of the attempt to recover, this method is invoked on
      the listener that was registered with the */
  public void onRecovery(NBRecoveryNotification recoveryNotification) {
    System.out.println(moduleName + recoveryNotification);
  }

  
  public void onEvent(NBEvent nbEvent) {
    String synopsis = (String) nbEvent.getContentSynopsis();
    System.out.println(moduleName + "Received NBEvent " + 
		       new String(nbEvent.getContentPayload()) ); 
    
    try {
      processEventProperties(nbEvent);
    } catch (NBEventException nbe) {
      System.out.println(moduleName + nbe);
    }

  }

  public void processEventProperties(NBEvent nbEvent) throws NBEventException {
    if (nbEvent.hasEventProperties()) {
      System.out.println(moduleName + "The received event has properties: " + 
			 nbEvent.getEventProperties() );
      EventProperties eventProperties = nbEvent.getEventProperties();
      Enumeration propertyNames = eventProperties.getPropertyNames();
      
      while (propertyNames.hasMoreElements()) {
	Object _propertyName = propertyNames.nextElement();
	String propertyType = 
	  eventProperties.isMutable(_propertyName) ? "mutable":"immutable";
	System.out.println("Property [" + _propertyName + "] was last modified"
			   + " by " + 
			   eventProperties.getLastModifier(_propertyName) + 
			   ". This is a " + propertyType + " property");
	
      }
      
    }
    
    
  }
  
  public static void main(String[] args) {
    int entityId = 7007;
    if (args.length == 3) {
      entityId = Integer.parseInt(args[2]);
    }
    
    TDSClient client = new TDSClient(entityId);
    Properties props = new Properties();
    String module = "TDSClient.main() ->";
    /** These properties pertain to setting up a TCP link */
    props.put("hostname", args[0]); 
    props.put("portnum", args[1]);
    
 
    
    try { 
      client.initializeBrokerCommunications(props, "niotcp");
      client.initializeProducer();
      

      BufferedReader commandLine = 
	new BufferedReader(new InputStreamReader(System.in));
      int tracker = 0;
      while(true){
	String s = commandLine.readLine();
	if (s == null) {
	  System.out.println(module + "String is null!!!");
	  break;
	}
	
	if (s.equals("q") ) {
	  client.closeBrokerConnection();
	  System.out.print(module + "Closed Broker Connection");
	  break;
	}
	
	if (s.equals("u")) {
	  client.unsubscribe();
	}
	
	if (s.equals("s")) {
	  client.initializeConsumer();
	}

	if (s.equals("tds")) {
	  client.initializeConsumerWithTDS();
	}
	
	if (s.startsWith("r")) {
	  StringTokenizer tok = new StringTokenizer(s, " ");
	  tok.nextToken();
	  
	  long start = Long.parseLong(tok.nextToken());
	  long end = Long.parseLong(tok.nextToken());
	  
	  client.initiateReplay(start, end);
	}

	tracker++;
	client.publishEvent(s + "[" + tracker + "]" );
	//client.publishEvent(s + "[" + tracker + "]" );
      }/** end while-true */
      System.out.println(module + "exiting ");

    } catch (IOException ioe) {
       System.out.println(ioe);
    } catch (ServiceException serEx) {
      System.out.println(serEx);
    } catch (NBEventException nbe) {
      System.out.println(nbe);
    }

    ThreadGroup tg = Thread.currentThread().getThreadGroup();
    int activeCount = tg.activeCount();
    if (activeCount != 0) {
      Thread[] threads = new Thread[activeCount];
      tg.enumerate(threads);

      for (int i=0; i < activeCount; i++) {
	System.out.println(threads[i].getName());
      }/* end for */

    }/* end if */
    
  }
}









